package com.xiaohu.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo16Cache {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("缓存演示")
    val sc: SparkContext = new SparkContext(conf)
    //===================================================================

    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    val studentsRDD: RDD[Student2] = linesRDD.map(_.split(","))
      .map {
        case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
          Student2(id, name, age.toInt, gender, clazz)
      }

    /**
     * 缓存：
     * 缓存的目的是为了spark core作业执行的时候，缩短rdd的执行链，能够更快的得到结果
     * 缓存的实现方式：
     *  1、需要缓存的rdd调用cache函数
     *  2、persist(StorageLevel.MEMORY_ONLY) 修改缓存级别
     *
     */
//    studentsRDD.cache() // 默认将rdd缓存到内存中，缓存级别为memory_only
//    studentsRDD.persist(StorageLevel.MEMORY_AND_DISK)

    //需求1：求每个班级的人数
    val rdd1: RDD[(String, Iterable[Student2])] = studentsRDD.groupBy(_.clazz)
    val resRDD1: RDD[(String, Int)] = rdd1.map((kv: (String, Iterable[Student2])) => (kv._1, kv._2.size))
    resRDD1.foreach(println)

    //需求2：求每个年龄的人数
    val rdd2: RDD[(Int, Iterable[Student2])] = studentsRDD.groupBy(_.age)
    val resRDD2: RDD[(Int, Int)] = rdd2.map((kv: (Int, Iterable[Student2])) => (kv._1, kv._2.size))
    resRDD2.foreach(println)

    while (true){

    }


  }
}

case class Student2(id:String,name:String,age:Int,gender:String,clazz:String)